#include "Connection.h"
#include "ConnectionListener.h"


Connection::Connection(boost::asio::io_service& io_service,
					   ConnectionClosedListener* closed_listener,
					   ConnectionListener* listerner): 
					   socket_(io_service),
					   m_closed_listener(closed_listener),
					   m_listener(listerner),
					   writting_(false),
					   closed_(false),
					   m_incoming(new ConstBufferSequence),
					   m_outcoming(new ConstBufferSequence)
{
	LogFunctionEnter;
}

Connection:: ~Connection()
{
	LogFunctionEnter;
}

boost::asio::ip::tcp::socket& Connection::socket()
{
	return socket_;
}

bool Connection::start()
{
	LogFunctionEnter;

	if (!socket_.is_open())
	{
		LOG_WARNING << "Connection start but close!";
		return false;
	}

	socket_.async_read_some(boost::asio::buffer(buffer_),
		boost::bind(&Connection::handleRead, shared_from_this(),
		boost::asio::placeholders::error,
		boost::asio::placeholders::bytes_transferred));

	onStart();

	if (m_closed_listener)
		m_closed_listener->onConnClosed(shared_from_this());

	if (m_listener)
		m_listener->onConnStart(shared_from_this());

	return true;
}

void Connection::stop(bool isNotify)
{
	LogFunctionEnter;

	writting_ = false;

	if (!closed_)
	{
		socket_.close();
		closed_ = true;

		onStop();

		if (isNotify && m_listener)
			m_listener->onConnStop(shared_from_this());
	}
}

void Connection::handleRead(const boost::system::error_code& e,
							std::size_t bytes_transferred)
{
	LogFunctionEnter;
	if (closed_)
		return;

	if (!e)
	{
		onReadRawData(buffer_.data(), bytes_transferred);

		if (m_listener)
			m_listener->onConnReadRawData(shared_from_this(),buffer_.data(), bytes_transferred);

		socket_.async_read_some(boost::asio::buffer(buffer_),
			boost::bind(&Connection::handleRead, shared_from_this(),
			boost::asio::placeholders::error,
			boost::asio::placeholders::bytes_transferred));

	}
	else if (e != boost::asio::error::operation_aborted)
	{
		stop();
		/*connection_manager_.stop(shared_from_this());*/
	}
}

void Connection::pushData(ArrayBuffer* buffer)
{
	LogFunctionEnter;
// 	if (closed_)
// 		return;

	onPushRawData(buffer->data(),buffer->size());

	if (m_listener)
		m_listener->onConnPushRawData(shared_from_this(),buffer->data(),buffer->size());

	boost::lock_guard<boost::mutex> lg(mutex_);
	m_incoming->pushData(buffer);

}

void Connection::startWrite()
{
	//LogFunctionEnter;
	if (closed_)
		return;

	if (!writting_)
	{
		{
			boost::lock_guard<boost::mutex> lg(mutex_);
			boost::swap(m_incoming,m_outcoming);
		}
		if (!m_outcoming->empty())
		{
			writting_ = true;
			boost::asio::async_write(socket_,*m_outcoming.get(), 
				boost::bind(&Connection::handleWrite, shared_from_this(),
				boost::asio::placeholders::error));
		}
		else
		{
			writting_ = false;
		}
	} 
}

void Connection::handleWrite(const boost::system::error_code& e)
{
	LogFunctionEnter;
	if (closed_)
		return;

	if (!e)
	{
		m_outcoming->clear();

		if (writting_)
		{
			{
				boost::lock_guard<boost::mutex> lg(mutex_);
				boost::swap(m_incoming,m_outcoming);
			}
			if (!m_outcoming->empty())
			{
				boost::asio::async_write(socket_,*m_outcoming.get(), 
					boost::bind(&Connection::handleWrite, shared_from_this(),
					boost::asio::placeholders::error));
			}
			else
			{
				writting_ = false;
			}
		}

	} 
	else if (e != boost::asio::error::operation_aborted)
	{
		stop();
		//connection_manager_.stop(shared_from_this());
	}
}


void Connection::update(uint32 timeElapse)
{
	if (isStopped())
		return;

	Runable::update(timeElapse);
}
